package com.apex.flink.stream;

import com.apex.env.Execution;
import com.apex.flink.FlinkEnvironment;
import com.apex.flink.utils.TableUtil;
import com.apex.plugin.BasePlugin;
import com.typesafe.config.Config;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSink;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

import java.util.ArrayList;
import java.util.List;
import java.util.Objects;

@SuppressWarnings("rawtypes")
public class FlinkStreamExecution implements Execution<FlinkStreamSource,FlinkStreamTransform,FlinkStreamSink> {

    private final FlinkEnvironment flinkEnvironment;

    private Config config;
    //注册的源表,在DataStream中也表现为注册维表关联
    private String tableName;
    //临时转换数据集
    private final String resultTableName = "apexosft_resutl_table";
    //保存执行的SQL脚本文件
    private List<String> sqlAll;

    public FlinkStreamExecution(FlinkEnvironment flinkEnvironment) {
        this.flinkEnvironment = flinkEnvironment;
    }

    /**
     * 程序处理方法
     * @param sources 数据源
     * @param transforms 业务处理
     * @param sinks 写入目标数据
     */
    @Override
    public void start(List<FlinkStreamSource> sources, List<FlinkStreamTransform> transforms, List<FlinkStreamSink> sinks) {
        tableName = flinkEnvironment.getQueryTable();
        List<DataStream> data = new ArrayList<>();
        DataStreamSink<Row> result = null;
        DataStream dataStreamTransform = null;
        for (FlinkStreamSource source : sources) {
            DataStream dataSet = source.getData(flinkEnvironment);
            data.add(dataSet);
            if (!StringUtils.isBlank(tableName)){
                registerSourceTable(source,dataSet);
            }
        }

        DataStream input = data.get(0);
        //SQL脚本模式判断
        DataStream sqlStream = null;

        for (FlinkStreamTransform transform : transforms) {
            if (!Objects.isNull(input)) {
                dataStreamTransform = fromSourceTable(transform);
                input = transform.processStream(flinkEnvironment, dataStreamTransform);
                registerResultTable(transform, input);
            }else {
                //启动SQL脚本模式
                sqlStream = transform.processStream(flinkEnvironment,null);
            }
        }

        for (FlinkStreamSink sink : sinks) {
            if (!Objects.isNull(dataStreamTransform)){
                DataStream dataSet = fromResultTable(sink);
                result = sink.outputStream(flinkEnvironment, dataSet);
            }else if (!Objects.isNull(sqlStream)){
                result = sink.outputStream(flinkEnvironment,sqlStream);
            }else if (!Objects.isNull(input)){
                result = sink.outputStream(flinkEnvironment,input);
            }
        }
        try {
            if (!Objects.isNull(result)) {
                //执行DataStream模式
                flinkEnvironment.getStreamExecutionEnvironment().execute(flinkEnvironment.getJobName());
            }else if (Objects.isNull(sqlStream)){
                //执行SQL脚本模式
                flinkEnvironment.getStreamTableEnvironment().execute(flinkEnvironment.getJobName());
            }
        } catch (Exception e) {
            e.printStackTrace();
        }

    }

    @Override
    public boolean status() {
        return false;
    }


    @Override
    public void prepare(Void plugin) {

    }

    @Override
    public Config getConfig() {
        return config;
    }

    @Override
    public void setConfig(Config config) {
        this.config = config;
    }

    /**
     * 将dataStream注册为一张可以查询的结果表
     * @param plugin 判断用
     * @param dataStream 数据集
     */
    private void registerSourceTable(BasePlugin plugin, DataStream dataStream) {
        StreamTableEnvironment tableEnvironment = flinkEnvironment.getStreamTableEnvironment();
        if (!TableUtil.tableExists(tableEnvironment, tableName)) {
            tableEnvironment.createTemporaryView(tableName,
                    dataStream);
        }
    }

    /**
     * 注册结果集临时表用于转换
     * @param plugin  判断用
     * @param dataStream 数据集
     */
    private void registerResultTable(BasePlugin plugin, DataStream dataStream) {
        StreamTableEnvironment tableEnvironment = flinkEnvironment.getStreamTableEnvironment();
        if (!TableUtil.tableExists(tableEnvironment, resultTableName)) {
            tableEnvironment.createTemporaryView(resultTableName,
                    dataStream);
        }
    }

    /**
     * 将数据集转换为一个dataStream供transform阶段额外的数据处理使用
     * @param plugin 判断用
     * @return 数据集
     */
    private DataStream<Row> fromSourceTable(BasePlugin plugin) {
        StreamTableEnvironment tableEnvironment = flinkEnvironment.getStreamTableEnvironment();
        if (StringUtils.isBlank(tableName)){
            return null;
        }else {
            Table table = tableEnvironment.from(tableName);
            return TableUtil.tableToDataStream(tableEnvironment, table,false);
        }
    }

    /**
     * 将数据集转换为一个dataStream供sink阶段输出使用
     * @param plugin 判断用
     * @return DataStream<Row>
     */
    private DataStream<Row> fromResultTable(BasePlugin plugin) {
        StreamTableEnvironment tableEnvironment = flinkEnvironment.getStreamTableEnvironment();
        Table table = tableEnvironment.from(resultTableName);
        return TableUtil.tableToDataStream(tableEnvironment, table,false);
    }
    /**
     * 执行的SQL文件
     * @param sqlAll SQL文件数据集
     */
    public void setSqlAll(List<String> sqlAll){
        this.sqlAll = sqlAll;
    }

    public List<String> getSqlAll(){
        if (Objects.isNull(sqlAll)){
            sqlAll = new ArrayList<>();
        }
        return sqlAll;
    }
}
